package data_manipulate;

/*
original text data be like:
We're gonna say it: Robert Allen Zimmerman is arguably the bard in modern times, you only have to look at the Bob Dylan songs and their lyrics to see this.

UPDATE: Bob Dylan is now 81 years old. He celebrates his birthday on 24 May and there are no signs that he is slowing down. Here we chart 50 great pieces of wisdom from some of the best Bob Dylan albums.

The artist otherwise known as Bob Dylan - Nobel Prize-winning no less - has been responsible for some of the finest pieces of prose and poetry of the last century, and set them to some of the most culturally important songs that there have ever been. (Don't mean to drag you into this, Ed Sheeran, but you've a way to go yet mate.)

Dylan has done all this while remaining an idiosyncratic - and let's face it, curmudgeonly - character, prone to wild left-turns in style. Put simply: he'll chase any musical whim he fancies (see 1965's Bringing It All Back Home, where he abandoned acoustic instruments in favour of electric ones and Street Rock, his 1986 collaboration with hip-hop pioneer Kurtis Blow).

In celebration of the man, we’ve picked our 50 favourite bits of Dylan's lyrical wisdom.

 */


import Utils.FileUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountSort {

    public static String inputPath = "hdfs://localhost:9000/sort/wordcountsort";
    public static String outputPath = "hdfs://localhost:9000/sort/output_wc";

    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        FileUtil fileUtil = new FileUtil(conf, inputPath);
        fileUtil.doError(inputPath);
        fileUtil.doDelete(outputPath);

        Job job = Job.getInstance(conf, "word count sort");
        job.setJarByClass(WordCountSort.class);

        job.setMapperClass(doMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(doReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        //job.setGroupingComparatorClass(doReduceComparator.class);

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

    public static class doMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        private final IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().trim().split(" ");
            for (String split : splits) {
                word.set(split);
                context.write(word, one);
            }

        }
    }

    public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        private int sum;
        private IntWritable ttl = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            ttl.set(sum);
            context.write(key, ttl);
        }
    }


    public static class doReduceComparator extends WritableComparator {
        public doReduceComparator() {
            super(IntWritable.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            return -super.compare(a, b);
        }
    }

}
